package com.bbx.flink.demo.data_stream_state.functions;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class ListStateDemo extends RichFlatMapFunction<Integer, Integer> {
    private static final long serialVersionUID = -3118729883975176402L;

    private ListState<Integer> listState;


    @Override
    public void flatMap(Integer value, Collector<Integer> out) throws Exception {
        listState.add(value);
        Iterable<Integer> integers = listState.get();
        integers.forEach(i->out.collect(i));

    }

    @Override
    public void open(Configuration parameters) throws Exception {
         listState = getRuntimeContext().getListState(new ListStateDescriptor("listState", Integer.class));
    }
}
